package com.uuah.server.transport.provider.mina;

import java.net.InetSocketAddress;
import java.net.URI;
import java.util.Map;
import java.util.concurrent.Executor;

import javax.management.ObjectName;
import javax.resource.spi.work.Work;
import javax.resource.spi.work.WorkException;
import javax.resource.spi.work.WorkManager;

import org.apache.mina.core.service.IoAcceptor;
import org.apache.mina.core.service.IoHandler;
import org.apache.mina.filter.codec.ProtocolCodecFactory;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.logging.MdcInjectionFilter;
import org.apache.mina.integration.jmx.IoFilterMBean;
import org.apache.mina.integration.jmx.IoServiceMBean;

import com.uuah.exception.UuahException;
import com.uuah.exception.lifecycle.CreateException;
import com.uuah.exception.lifecycle.DisposeException;
import com.uuah.exception.lifecycle.InitialisationException;
import com.uuah.server.UuahServer;
import com.uuah.server.api.component.service.IService;
import com.uuah.server.api.endpoint.IUuahEndpoint;
import com.uuah.server.api.transport.IConnector;
import com.uuah.server.api.transport.retry.IRetryCallback;
import com.uuah.server.api.transport.retry.IRetryContext;
import com.uuah.server.config.i18n.ServerMessages;
import com.uuah.server.context.notification.AgentMBeanNotification;
import com.uuah.server.exception.transport.ConnectException;
import com.uuah.server.transport.AbstractMessageReceiver;
import com.uuah.server.transport.provider.mina.codecs.UuahMessageCodecFactory;
import com.uuah.server.transport.provider.mina.filter.MinaExecutorFilter;
import com.uuah.server.transport.provider.mina.filter.WhitelistFilter;
import com.uuah.utils.ClassUtils;

/**
 *
 * TODO comment ...
 *
 * <p>
 * update record:updator，update time，update content and version
 * </p>
 *
 * @author <a href="jonny_quan@hotmail.com">jonny</a>
 * @date 2009 2009-3-27 下午02:45:59
 * @version 1.0.0
 */
public class MinaMessageReceiver extends AbstractMessageReceiver implements
		Work {
	protected IoAcceptor serverAcceptor;

	public IoAcceptor getServerAcceptor() {
		return serverAcceptor;
	}

	@Override
	public void doConnect() throws Exception {
		disposing.set(false);
		URI uri = endpoint.getEndpointURI().getUri();
		try {
			serverAcceptor = ((MinaConnector) connector).getServerAcceptor(uri);
		} catch (Exception e) {
			throw new com.uuah.server.exception.transport.ConnectException(
					ServerMessages.failedToBindToUri(uri), e, this);
		}

		connector.fireNotification(new AgentMBeanNotification(
				new IoServiceMBean(serverAcceptor), new ObjectName(
						"com.uuah.server.transport.mina:type=serverAcceptor,name="
								+ service),
				AgentMBeanNotification.CREATE_MBEAN_ACTION));
		try {
			getWorkManager().scheduleWork(this, WorkManager.INDEFINITE, null,
					connector);
		} catch (WorkException e) {
			throw new ConnectException(ServerMessages.failedToScheduleWork(),
					e, this);
		}
	}

	public void run() {
		if (connector.isStarted() && !disposing.get()) {
			try {
				retryTemplate.execute(new IRetryCallback() {
					public void doWork(IRetryContext context) throws Exception {
						try {
							// "handlerClass","remoteAddress"
							// "localAddress"
							if (endpoint.isIpFilter()) {
								WhitelistFilter whiteListFilter = new WhitelistFilter();
								// whiteListFilter.setWhilelist(addresses)
								serverAcceptor.getFilterChain().addLast(
										"whiteListFilter", whiteListFilter);

								connector
										.fireNotification(new AgentMBeanNotification(
												new IoFilterMBean(
														whiteListFilter),
												new ObjectName(
														"com.uuah.server.transport.mina:type=filter,name=whiteListFilter["
																+ service + "]"),
												AgentMBeanNotification.CREATE_MBEAN_ACTION));
							}
							MdcInjectionFilter mdcInjectionFilter = new MdcInjectionFilter();
							serverAcceptor.getFilterChain().addLast("mdc",
									mdcInjectionFilter);
							ProtocolCodecFactory protocolFactory = getProtocolCodecFactory();
							if (protocolFactory != null) {
								serverAcceptor.getFilterChain()
										.addLast(
												"protocolFilter",
												new ProtocolCodecFilter(
														protocolFactory));
							}

							Executor executor = (Executor) connector
									.getReceiverThreadingProfile().createPool(
											"MinaProtocolHandler");

							UuahServer.getServerStats().setMinaThreadPoolSize(
									connector.getReceiverThreadingProfile()
											.getMaxThreadsActive());
							MinaExecutorFilter executorFilter = new MinaExecutorFilter(
									executor);
							serverAcceptor.getFilterChain().addLast("Executor",
									executorFilter);

							connector
									.fireNotification(new AgentMBeanNotification(
											new IoFilterMBean(executorFilter),
											new ObjectName(
													"com.uuah.server.transport.mina:type=filter,name=executorFilter["
															+ service + "]"),
											AgentMBeanNotification.CREATE_MBEAN_ACTION));
							serverAcceptor.setHandler(getServerHandler());
							serverAcceptor.bind();
							if (logger.isInfoEnabled()) {
								logger.warn("Server now listening on port:"
										+ ((InetSocketAddress) serverAcceptor
												.getDefaultLocalAddress())
												.getPort());
							}

						} catch (Exception ex) {
							if (!connector.isDisposed() && !disposing.get()) {
								throw new ConnectException(ex, null);
							}
						}
					}

					public String getWorkDescription() {
						return getConnectionDescription();
					}
				});
			} catch (Exception e) {
				handleException(e);
			}
		}

	}

	public MinaMessageReceiver(IConnector connector, IService service,
			IUuahEndpoint endpoint) throws InitialisationException,
			CreateException {
		super(connector, service, endpoint);

	}

	@Override
	public void doDisconnect() throws Exception {
		disposing.set(true);
		if (serverAcceptor != null) {
			if (logger.isDebugEnabled()) {
				logger.debug("Unbinding: " + serverAcceptor);
			}
			serverAcceptor.unbind();
		}
	}

	@Override
	protected void doDispose() {
		try {
			if (serverAcceptor != null && serverAcceptor.isActive()) {
				if (logger.isDebugEnabled()) {
					logger.debug("Closing: " + serverAcceptor);
				}
				serverAcceptor.unbind();
				serverAcceptor.dispose();
			}
			serverAcceptor = null;
		} catch (Exception e) {
			logger.error(new DisposeException(ServerMessages
					.failedToCloseSocket(), e, this).getMessage());
		}
		if (logger.isDebugEnabled()) {
			logger.debug("Closed Mina Port!");
		}
	}

	@Override
	protected void doStart() throws UuahException {
	}

	@Override
	protected void doStop() throws UuahException {
		// template method
	}

	public void release() {
		// template method
	}

	@SuppressWarnings("unchecked")
	protected ProtocolCodecFactory getProtocolCodecFactory() throws Exception {
		Map<String, String> props = endpoint.getProperties();
		String protocolFactoryClassName = (String) props.get("protocolFactory");
		if (protocolFactoryClassName == null) {
			if (logger.isDebugEnabled()) {
				logger.debug("Using default protocol codecs");
			}
			protocolFactoryClassName = UuahMessageCodecFactory.class.getName();
		}

		return (ProtocolCodecFactory) ClassUtils.instanciateClass(
				protocolFactoryClassName, new Object[] { true });
	}

	protected IoHandler getServerHandler() throws UuahException {
		return new MinaServerHandler(connector, service, endpoint);
	}
}